package com.example;

import com.example.domain.UserEvent;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Ann
 * @version v1.0
 * @ClassName StatefulPractice
 * @Description
 * @date Created in 2025/5/20
 **/
public class StatefulPractice {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Simulate input stream
        DataStream<UserEvent> stream = env.fromElements(
                new UserEvent("A", 100, 1000L),
                new UserEvent("B", 200, 2000L),
                new UserEvent("A", 150, 4000L),
                new UserEvent("B", 300, 6000L)
        );

        stream
//                .keyBy(new KeySelector<UserEvent, String>() {
//
//                    @Override
//                    public String getKey(UserEvent userEvent) throws Exception {
//                        return userEvent.userId;
//                    }
//                })
                .keyBy(userEvent -> userEvent.userId)
                .map(new RichMapFunction<UserEvent, Object>() {
                    ValueState<Integer> count;

                    @Override
                    public void open(Configuration parameters) throws Exception {
//                        count = getRuntimeContext().getState();
                    }


                    public Object map(UserEvent userEvent) throws Exception {
//                        Integer currentCount = count.value();
//                        if (currentCount == null) {
//                            currentCount = 0;
//                        }
//                        count.update(currentCount + 1);
                        return null;
                    }
                });
    }
}
